package com.ygl.hbase;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ygl.hbase.pojo.*;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Date 2022/1/24 13:28
 * @Created yanggld
 * @Description
 */
public class HbaseExample {
    private static  HbaseUtil hbaseUtil;
    public static void main(String[] args) throws Exception {
        hbaseUtil = new HbaseUtil("mh1,mh2,mh3");
        boolean tableExists = hbaseUtil.tableExists("student1");
        System.out.println("table exists===>:"+tableExists);

//        System.out.println("stu2:"+hbaseUtil.tableExists("stu2"));
//        hbaseUtil.createTable("stu2","info");
//        hbaseUtil.dropTable("stu2");
//        System.out.println("stu2:"+hbaseUtil.tableExists("stu2"));

//        hbaseUtil.createNamespace("namespace");
//        hbaseUtil.dropTable("namespace:stu2");
//        hbaseUtil.createTable("namespace:stu2","info","info1");
//        hbaseUtil.putData("namespace:stu2","stu001","info","name","里斯");
//        hbaseUtil.putData("namespace:stu2","stu001","info","age","18");
//        hbaseUtil.putData("namespace:stu2","stu001","info1","age","28");
//        hbaseUtil.putData("namespace:stu2","stu001","info1","sex","male");
//        hbaseUtil.putData("namespace:stu2","stu002","info","name","王五");
//        hbaseUtil.putData("namespace:stu2","stu002","info","age","88");
//        hbaseUtil.putData("namespace:stu2","stu003","info","name","aifosen");
//        hbaseUtil.putData("namespace:stu2","stu003","info","age","38");

//
//        System.out.println("----------------------");
//        HBaseDBO hBaseDBO = hbaseUtil.getData("namespace:stu2", "stu001");
////        System.out.println(hBaseDBO.toString());
//        {
//            String rowKey = hBaseDBO.getRowKey();
//            List<HbaseFamily> familyList = hBaseDBO.getCfList();
//            for (HbaseFamily hbaseFamily : familyList) {
//                String family = hbaseFamily.getCf();
//                Map<String, Object> data = hbaseFamily.getData();
//                System.out.println(rowKey+":"+family+":"+data);
//            }
//        }
//
//        System.out.println("----------------------");
//        HBaseDBOByCf hBaseDBO2 = hbaseUtil.getData("namespace:stu2", "stu001", "info");
//        {
//            String rowKey = hBaseDBO2.getRowKey();
//            String cf = hBaseDBO2.getCf();
//            Map<String, Object> data = hBaseDBO2.getData();
//            System.out.println(rowKey+":"+cf+":"+data);
//        }
//
//        System.out.println("----------------------");
//        HBaseDBOByCfCn hBaseDBO3 = hbaseUtil.getData("namespace:stu2", "stu001", "info","name");
//        {
//            String rowKey = hBaseDBO3.getRowKey();
//            String cf = hBaseDBO3.getCf();
//            String cn = hBaseDBO3.getCn();
//            Object value = hBaseDBO3.getValue();
//            System.out.println(rowKey+":"+cf+":"+cn+":"+value);
//        }
//        hbaseUtil.putData("namespace:stu2","1000","info","age","1000x");
//        hbaseUtil.putData("namespace:stu2","1000","info1","age","1000y");
//        hbaseUtil.putData("namespace:stu2","1001","info","age","1001x");
//        hbaseUtil.putData("namespace:stu2","1001","info1","age","1001y");
//        hbaseUtil.putData("namespace:stu2","1002","info1","name","1002x");
//        hbaseUtil.putData("namespace:stu2","1002","info1","age","1002y");
//
//        System.out.println("getDataList----------------------");
//        List<HBaseDBO> dataList = hbaseUtil.getDataList("namespace:stu2");
//        for (HBaseDBO dbo : dataList) {
//            String rowKey = dbo.getRowKey();
//            List<HbaseFamily> familyList = dbo.getCfList();
//            for (HbaseFamily hbaseFamily : familyList) {
//                String family = hbaseFamily.getCf();
//                Map<String, Object> data = hbaseFamily.getData();
//                System.out.println(rowKey+":"+family+":"+data);
//            }
//        }
//
//        System.out.println("getDataList2----------------------");
//        List<HBaseDBO> dataList2 = hbaseUtil.getDataList("namespace:stu2","stu001",true,"stu003",true);
//        for (HBaseDBO dbo : dataList2) {
//            String rowKey = dbo.getRowKey();
//            List<HbaseFamily> familyList = dbo.getCfList();
//            for (HbaseFamily hbaseFamily : familyList) {
//                String family = hbaseFamily.getCf();
//                Map<String, Object> data = hbaseFamily.getData();
//                System.out.println(rowKey+":"+family+":"+data);
//            }
//        }
//
//        hbaseUtil.deleteData("namespace:stu2","1000");
//        hbaseUtil.deleteData("namespace:stu2","1001","info");
//        hbaseUtil.deleteData("namespace:stu2","1002","info1","age");
//
//        System.out.println("getDataList----------------------");
//        List<HBaseDBO> dataList3 = hbaseUtil.getDataList("namespace:stu2");
//        for (HBaseDBO dbo : dataList3) {
//            String rowKey = dbo.getRowKey();
//            List<HbaseFamily> familyList = dbo.getCfList();
//            for (HbaseFamily hbaseFamily : familyList) {
//                String family = hbaseFamily.getCf();
//                Map<String, Object> data = hbaseFamily.getData();
//                System.out.println(rowKey+":"+family+":"+data);
//            }
//        }


        /**
         * 常用功能测试
         * 1. 新增
         * 2. 批量新增
         * 3. 通过id / ids查询
         * 4. 通过条件模糊查询
         * 5. 分页查询
         * 6. 条件模糊分页查询
         * 7. 修改
         * 8. 删除单个
         * 9. 批量删除
         */
//        System.err.println("listTableNames---------------");
//        listTableNames();
//        if (hbaseUtil.tableExists("person")) {
//            hbaseUtil.dropTable("person");
//        }
//        if (hbaseUtil.tableExists("animal")) {
//            hbaseUtil.dropTable("animal");
//        }
//        listTableNames();
//        hbaseUtil.createTable("person","info","info1");
//        hbaseUtil.truncateTable("person",true);
//        hbaseUtil.createTable("animal","info","info1");
//        hbaseUtil.truncateTable("animal",true);
//        listTableNames();
//        System.err.println("insertOne---------------");
//        // 查询所有
//        getAll();
//        //1. 新增单表数据
//        insertOne();
//        // 3. 通过id / ids查询
//        getByIds();
//        System.err.println("deleteById---------------");
//        // 8. 删除
//        deleteById();
//        getAll();
//        System.err.println("binsert---------------");
//        //2. 批量新增
//        binsert();
//        getAll();
//        System.err.println("bdelete---------------");
//        // 9. 批量删除单表
//        bdelete();
//        getAll();
//        System.err.println("bdeletes---------------");
//        // 9. 批量删除多表
//        bdeletes();
//        getAll();
//        System.err.println("update---------------");
//        //1. 新增单表数据
//        insertOne();
//        getByIds();
//        // 7. 修改
//        update();
//        getAll();
             /* 行键过滤器 RowFilter
             * 列族过滤器 FamilyFilter
             * 列名过滤器 QualifierFilter
             * 值过滤器 ValueFilter
             * 参考列过滤器 DependentColumnFilter
             * 单列值过滤器 SingleColumnValueFilter
             * 单列值排除器 SingleColumnValueExcludeFilter
             * 行前缀过滤器 PrefixFilter
             * 列前缀过滤器 ColumnPrefixFilter
             * 分页过滤器 PageFilter
             * 行键过滤器 KeyOnlyFilter
             * 首次行键过滤器 FirstKeyOnlyFilter
             * 包含结束的过滤器 InclusiveStopFilter
             * 时间戳过滤器 TimestampsFilter
             * 列计数过滤器 ColumnCountGetFilter
             * 列分页过滤器 ColumnPaginationFilter
             * 跳转过滤器 SkipFilter
             * 全匹配过滤 WhileMatchFilter
             * 多种过滤条件的使用方法
             */
//        DependentColumnFilter dependentColumnFilter = new DependentColumnFilter();
//        hbaseUtil.truncateTable("person",true);
//        binsertTest();

        System.out.println("多个filter一起工作,所有filter都满足或者满足其中一个");
        SingleColumnValueFilter filter1 = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("name"), CompareOperator.EQUAL, new BinaryPrefixComparator(Bytes.toBytes("name1_1")));
        SingleColumnValueFilter filter2 = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("name"), CompareOperator.NOT_EQUAL, new BinaryComparator(Bytes.toBytes("name1_10")));
        List<Filter> filters = new ArrayList<>();
        filters.add(filter1);
        filters.add(filter2);
        Filter filter = new FilterList(FilterList.Operator.MUST_PASS_ALL, filters);
        HBaseModel p15 = hbaseUtil.filter("person", filter);
        System.out.println(new ObjectMapper().writeValueAsString(p15));

        System.out.println("查询遇到某种条件数据之前的数据。当遇到不符合设定条件的数据的时候，整个扫描也就结束了");
        Filter whileMatchFilter = new WhileMatchFilter(new ValueFilter(CompareOperator.NOT_EQUAL, new BinaryComparator(Bytes.toBytes("name1_2"))));
        HBaseModel p14 = hbaseUtil.filter("person", whileMatchFilter);
        System.out.println(new ObjectMapper().writeValueAsString(p14));

        System.out.println("所有列都符合条件的一行被留下。一行中的某一列不符合条件，那么整行就会被过滤掉");
        Filter skipFilter = new SkipFilter(new ValueFilter(CompareOperator.NOT_EQUAL, new BinaryComparator(Bytes.toBytes("name1_0"))));
        HBaseModel p13 = hbaseUtil.filter("person", skipFilter);
        System.out.println(new ObjectMapper().writeValueAsString(p13));

        System.out.println("返回所有数据，每条数据共limit列。每条数据从第offset列开始，共limit列。（总共列数等于每个列簇里面列数相加）");
        HBaseModel p12 = hbaseUtil.filter("person",  new ColumnPaginationFilter(2,1));
        System.out.println(new ObjectMapper().writeValueAsString(p12));

        System.out.println("每次查询pageSize条数据");
        HBaseModel p11 = hbaseUtil.filter("person", new PageFilter(10));
        System.out.println(new ObjectMapper().writeValueAsString(p11));

        System.out.println("某一条数据任一列名 开头是 目标数据，返回该条数据只是该列数据");
        HBaseModel p10 = hbaseUtil.filter("person", new ColumnPrefixFilter(Bytes.toBytes("name")));
        System.out.println(new ObjectMapper().writeValueAsString(p10));

        System.out.println("某一条数据rowkey 开头是 目标数据，返回该条数据全部列数据");
        HBaseModel p9 = hbaseUtil.filter("person", new PrefixFilter(Bytes.toBytes("1002")));
        System.out.println(new ObjectMapper().writeValueAsString(p9));

        System.out.println("某一条数据某一列值 匹配 目标数据，返回该条数据除了该列的所有列数据");
        HBaseModel p8 = hbaseUtil.filter("person", new SingleColumnValueExcludeFilter(Bytes.toBytes("info"),Bytes.toBytes("name"),CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("张三3"))));
        System.out.println(new ObjectMapper().writeValueAsString(p8));

        System.out.println("7. 某一条数据某一列值 匹配 目标数据，返回该条数据全部列数据");
        HBaseModel p7 = hbaseUtil.filter("person", new SingleColumnValueFilter(Bytes.toBytes("info"),Bytes.toBytes("name"),CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("name1_1"))));
        System.out.println(new ObjectMapper().writeValueAsString(p7));

        System.out.println("某一条数据任一列值 匹配 目标数据，返回该条数据只是该列数据");
        HBaseModel p5 = hbaseUtil.filter("person", new ValueFilter(CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("name1_0"))));
        System.out.println(new ObjectMapper().writeValueAsString(p5));

        System.out.println("某一条数据rowkey 匹配 目标数据，返回该条数据全部列数据");
        HBaseModel p = hbaseUtil.filter("person", new RowFilter(CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("1001"))));
        System.out.println(p);

        System.out.println("某一条数任一列族名 匹配 目标数据，返回该条数据只是该列簇下的全部列数据");
        HBaseModel p2 = hbaseUtil.filter("person", new FamilyFilter(CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("info1"))));
        System.out.println(new ObjectMapper().writeValueAsString(p2));

        System.out.println("某一条数任一列名 匹配 目标数据，返回该条数据只是该列数据");
        HBaseModel p3 = hbaseUtil.filter("person", new QualifierFilter(CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("age"))));
        System.out.println(new ObjectMapper().writeValueAsString(p3));




//        HBaseModel person2 = hbaseUtil.get("person", "1002");
//        System.out.println(person2);
    }

    public static void  listTableNames() throws IOException {
        System.out.println("----------------------");
        System.out.println(new ObjectMapper().writeValueAsString(hbaseUtil.listTableNames()));
    }


    // 8. 删除数据
    public static void deleteById() throws Exception {
        hbaseUtil.deleteData("person","1001");
    }

    // 9. 批量删除数据
    public static void bdelete() throws Exception {
        String text = "{\n" +
                "    \"tableName\":\"person\",\n" +
                "    \"rowKeys\":[\"1001\",\"1002\"]\n" +
                "}";
        HBaseGetModel getModel = new ObjectMapper().readValue(text, HBaseGetModel.class);
        hbaseUtil.deleteData(getModel);
    }
    // 9. 批量删除数据
    public static void bdeletes() throws Exception {
        String text2 = "[{\n" +
                "    \"tableName\":\"person\",\n" +
                "    \"rowKeys\":[\"1001\",\"1002\"]\n" +
                "},{\n" +
                "    \"tableName\":\"animal\",\n" +
                "    \"rowKeys\":[\"1001\",\"1002\"]\n" +
                "}]";
        List<HBaseGetModel> getModel2 = new ObjectMapper().readValue(text2, new TypeReference<List<HBaseGetModel>>() {
        });
        hbaseUtil.deleteData(getModel2);
    }



    //1. 新增单表数据
    public static void insertOne() throws Exception {
        String text = "{\n" +
                "    \"tableName\":\"person\",\n" +
                "    \"rows\":[\n" +
                "        {\n" +
                "            \"rowKey\":\"1001\",\n" +
                "            \"cf\":\"info\",\n" +
                "            \"name\":\"张三\",\n" +
                "            \"age\":\"3\"\n" +
                "        },\n" +
                "        {\n" +
                "            \"rowKey\":\"1001\",\n" +
                "            \"cf\":\"info1\",\n" +
                "            \"name\":\"张四\",\n" +
                "            \"age\":\"4\"\n" +
                "        },\n" +
                "        {\n" +
                "            \"rowKey\":\"1002\",\n" +
                "            \"cf\":\"info\",\n" +
                "            \"name\":\"王五\",\n" +
                "            \"age\":\"5\"\n" +
                "        },\n" +
                "        {\n" +
                "            \"rowKey\":\"1002\",\n" +
                "            \"cf\":\"info1\",\n" +
                "            \"name\":\"王六\",\n" +
                "            \"age\":\"6\"\n" +
                "        }\n" +
                "    ]\n" +
                "}";
        HBaseModel model = new ObjectMapper().readValue(text, HBaseModel.class);
        hbaseUtil.putData(model);
    }

    public static void binsertTest()throws Exception {
        List<HBaseModel> models = new ArrayList<>();
        int count = 40;
        for (int i = 0; i < count; i++) {
            HBaseModel model = new HBaseModel();
            model.setTableName("person");
            List<Map<String,String>> rows = new ArrayList<>();
            Map<String,String> map1 = new HashMap<>();
            map1.put("rowKey","rowKey_"+i);
            map1.put("cf","info");
            map1.put("name","name1_"+i);
            map1.put("age","age1_"+i);
            map1.put("sex","sex1_"+i);
            rows.add(map1);
            Map<String,String> map2 = new HashMap<>();
            map2.put("rowKey","rowKey_"+i);
            map2.put("cf","info1");
            map2.put("name","name2_"+i);
            map2.put("age","age2_"+i);
            map2.put("sex","sex2_"+i);
            rows.add(map2);
            model.setRows(rows);
            models.add(model);
        }
        hbaseUtil.putData(models);
    }

    //2. 批量新增
    public static void binsert() throws Exception {
        String text = "[{\n" +
                "    \"tableName\":\"person\",\n" +
                "    \"rows\":[\n" +
                "        {\n" +
                "            \"rowKey\":\"1001\",\n" +
                "            \"cf\":\"info\",\n" +
                "            \"name\":\"张三\",\n" +
                "            \"age\":13\n" +
                "        },\n" +
                "        {\n" +
                "            \"rowKey\":\"1002\",\n" +
                "            \"cf\":\"info\",\n" +
                "            \"name\":\"张四\",\n" +
                "            \"age\":14\n" +
                "        },\n" +
                "{\n" +
                "            \"rowKey\":\"1001\",\n" +
                "            \"cf\":\"info1\",\n" +
                "            \"name\":\"王五\",\n" +
                "            \"age\":15\n" +
                "        },\n" +
                "{\n" +
                "            \"rowKey\":\"1001\",\n" +
                "            \"cf\":\"info1\",\n" +
                "            \"name\":\"王六\",\n" +
                "            \"age\":16\n" +
                "        }\n" +
                "    ]\n" +
                "},\n" +
                "{\n" +
                "    \"tableName\":\"animal\",\n" +
                "    \"rows\":[\n" +
                "        {\n" +
                "            \"rowKey\":\"1001\",\n" +
                "            \"cf\":\"info\",\n" +
                "            \"name\":\"狮子\",\n" +
                "            \"age\":\"8\"\n" +
                "        }\n" +
                "    ]\n" +
                "},\n" +
                "{\n" +
                "    \"tableName\":\"animal\",\n" +
                "    \"rows\":[\n" +
                "        {\n" +
                "            \"rowKey\":\"1002\",\n" +
                "            \"cf\":\"info\",\n" +
                "            \"name\":\"老鼠\",\n" +
                "            \"age\":\"3\"\n" +
                "        }\n" +
                "    ]\n" +
                "}\n" +
                "]";
        List<HBaseModel> models = new ObjectMapper().readValue(text, new TypeReference<List<HBaseModel>>() {
        });
        hbaseUtil.putData(models);
        HBaseModel model = hbaseUtil.getData("person", "1001");
        System.out.println("----------------------");
        System.out.println(new ObjectMapper().writeValueAsString(model));

        HBaseModel model2 = hbaseUtil.getData("person", "1001","info1");
        System.out.println("----------------------");
        System.out.println(new ObjectMapper().writeValueAsString(model2));

        HBaseModel model3 = hbaseUtil.getData("person", "1001","info1","name");
        System.out.println("----------------------");
        System.out.println(new ObjectMapper().writeValueAsString(model3));


        HBaseModel person2 = hbaseUtil.getDataList("person","1001",false,"1002",true);
        System.out.println("----------------------");
        System.out.println(new ObjectMapper().writeValueAsString(person2));
    }


    public static void getAll() throws Exception {
        HBaseModel person = hbaseUtil.getDataList("person");
        System.out.println("----------------------");
        System.out.println(new ObjectMapper().writeValueAsString(person));
    }


    // 3. 通过id / ids查询
    public static void getByIds() throws Exception {
        String text = "{\n" +
                "    \"tableName\":\"person\",\n" +
                "    \"rowKeys\":[\"1001\",\"1002\"]\n" +
                "}";
        HBaseGetModel getModel = new ObjectMapper().readValue(text, HBaseGetModel.class);
        HBaseModel dataList = hbaseUtil.getDataList(getModel);
        System.out.println("----------------------");
        System.out.println(new ObjectMapper().writeValueAsString(dataList));


        String text2 = "[{\n" +
                "    \"tableName\":\"person\",\n" +
                "    \"rowKeys\":[\"1001\",\"1002\"]\n" +
                "},{\n" +
                "    \"tableName\":\"animal\",\n" +
                "    \"rowKeys\":[\"1001\",\"1002\"]\n" +
                "}]";
        List<HBaseGetModel> getModel2 = new ObjectMapper().readValue(text2, new TypeReference<List<HBaseGetModel>>() {
        });
        List<HBaseModel> dataList2 = hbaseUtil.getDataList(getModel2);
        System.out.println("----------------------");
        System.out.println(new ObjectMapper().writeValueAsString(dataList2));
    }

    // 7. 修改
    public static void update() throws Exception {
        String text = "[{\n" +
                "    \"tableName\":\"person\",\n" +
                "    \"rows\":[\n" +
                "        {\n" +
                "            \"rowKey\":\"1001\",\n" +
                "            \"cf\":\"info\",\n" +
                "            \"name\":\"张三3\",\n" +
                "            \"age\":\"13\"\n" +
                "        },\n" +
                "        {\n" +
                "            \"rowKey\":\"1002\",\n" +
                "            \"cf\":\"info\",\n" +
                "            \"name\":\"张四4\",\n" +
                "            \"age\":\"14\"\n" +
                "        },\n" +
                "{\n" +
                "            \"rowKey\":\"1001\",\n" +
                "            \"cf\":\"info1\",\n" +
                "            \"name\":\"王五5\",\n" +
                "            \"age\":\"15\"\n" +
                "        },\n" +
                "{\n" +
                "            \"rowKey\":\"1001\",\n" +
                "            \"cf\":\"info1\",\n" +
                "            \"name\":\"王六6\",\n" +
                "            \"age\":\"16\"\n" +
                "        }\n" +
                "    ]\n" +
                "},\n" +
                "{\n" +
                "    \"tableName\":\"animal\",\n" +
                "    \"rows\":[\n" +
                "        {\n" +
                "            \"rowKey\":\"1001\",\n" +
                "            \"cf\":\"info\",\n" +
                "            \"name\":\"狮子2\",\n" +
                "            \"age\":\"80\"\n" +
                "        }\n" +
                "    ]\n" +
                "},\n" +
                "{\n" +
                "    \"tableName\":\"animal\",\n" +
                "    \"rows\":[\n" +
                "        {\n" +
                "            \"rowKey\":\"1002\",\n" +
                "            \"cf\":\"info\",\n" +
                "            \"name\":\"老虎\",\n" +
                "            \"age\":\"30\"\n" +
                "        }\n" +
                "    ]\n" +
                "}\n" +
                "]";
        List<HBaseModel> models = new ObjectMapper().readValue(text, new TypeReference<List<HBaseModel>>() {
        });
        hbaseUtil.putData(models);
    }
}
